// CSE571 - io-manager.cpp          
// Authors: Jonathan Wald, Guangle Fan
// Defines the active object used in the synchronous layer of the app.

#include "io-manager.h"

#include <iostream>
#include <sstream>
#include <iterator>

#include "async-io-svc-handler.h"
#include "isock_stream_iterator_T.h"
#include "osock_stream_iterator_T.h"
#include "globals.h"
#include "msg-flag.h"
#include "signal-handler.h"

const int TIMEOUT_VALUE = 2;

IOManager* IOManager::instance_ = NULL;

IOManager::IOManager (void) {
  shut_down = false;
}

IOManager* IOManager::instance() {
  if (instance_ == NULL) {
    instance_ = new IOManager();
  }
  return instance_;
}

int IOManager::open (int pool_size) {
  return this->activate (THR_NEW_LWP|THR_DETACHED, pool_size);
}

int IOManager::close (u_long) {
  //this->enqueue (0);
  ACE_OS::sleep (ACE_Time_Value (0, 250000));
  return(0);
}

int IOManager::enqueue (AsyncIOSvcHandler *handler, bool write)
{
  void *v_data = (void *) handler;
  char *c_data = (char *) v_data;

  ACE_Message_Block *mb;
  ACE_NEW_RETURN (mb,
                  ACE_Message_Block (c_data),
                  -1);
  // Put this message into our write map so that we know that this message
  // was a right when we pull it out of the queue later.  This helps 
  // us avoid a deadlock condition.
  if (write) {
    write_map.insert(mb);
  }
  if (this->putq (mb) == -1) {
    mb->release ();
    return -1;
  }
  return 0;
}

int IOManager::enqueue_resume(AsyncIOSvcHandler *handler) {
  void *v_data = (void *) handler;
  char *c_data = (char *) v_data;
  ACE_Message_Block *mb;
  ACE_NEW_RETURN (mb,
                  ACE_Message_Block (c_data),
                  -1);
  handler->resume_block = mb;
  if (this->putq (mb) == -1) {
    mb->release ();
    return -1;
  }
  return 0;
}

int IOManager::enqueue_suspend(AsyncIOSvcHandler *handler) {
  void *v_data = (void *) handler;
  char *c_data = (char *) v_data;
  ACE_Message_Block *mb;
  ACE_NEW_RETURN (mb,
                  ACE_Message_Block (c_data),
                  -1);
  handler->suspend_block = mb;
  if (this->putq (mb) == -1) {
    mb->release ();
    return -1;
  }
  return 0;
}



class Message_Block_Guard
{
public:
  Message_Block_Guard (ACE_Message_Block *&mb)
    : mb_ (mb)
  {
  }

  ~Message_Block_Guard (void)
  {
    mb_->release ();
  }

protected:
  ACE_Message_Block *&mb_;
};

int IOManager::svc (void) {
  ACE_Message_Block *mb;
  ACE_Time_Value tv(4);
  while (!SigHandler::sigint_fired()) {
    if(this->getq(mb, &tv) == -1) continue;  
    char *c_data = mb->base ();

    if (c_data) {
      void *v_data = (void *) c_data;
      AsyncIOSvcHandler *handler = (AsyncIOSvcHandler *) v_data;
      if (DEBUG) std::cout << "Dequeued and handling " << handler << std::endl;

      // If stream is closed and we are waiting for de-registration, continue.
      if(handler->counterpart_quit) continue;

      // If this was enqueued for in-band resume/suspend, process accordingly.
      if(handler->resume_block == mb) {
	handler->resume();
	mb->release();
	continue;
      } else if (handler->suspend_block ==mb) {
	handler->suspend();
	mb->release();
	continue;
      }

      // If blocks were new blocks were enqueued after we resumed,
      // ensure order by making them wait for suspended ones.
      if (handler->first_resume != NULL && mb != handler->first_resume) {
        this->putq(mb);
        continue;
      } else if (handler->first_resume != NULL && mb == handler->first_resume) {
        handler->first_resume = NULL;
      }

      // If this game is suspended, process accordingly.
      // Don't release mb is is reused by suspended.
      if(handler->suspended(mb)) {
	continue;
      }

      // Write_block will see if this handler was enqueued for writing
      // purposes and handle the writing if thats the case.  Otherwise, we are
      // reading.
      if (write_block (handler, mb)) {
        if (handler->last_act()) {
	  if (DEBUG) std::cout << "Writing act. " << handler << std::endl;
	  mb->release();
          break;
	}
	mb->release();
        continue;
      }

      // This method can be implemented by subclasses of async-io-svc-handler.
      // Used by dealer side players to delay a read if a game is not ready
      // to be restarted.
      if (!handler->ready_for_read()) {
        putq(mb, &tv);
        continue;
      }

      // Otherwise this enqueue was for a read.
      read_data = false;
      isock_stream_iterator<char> begin_iter (handler->peer());
      isock_stream_iterator<char> end_iter;
      
      // Accumulate our incoming data in a string stream.
      std::stringstream stream;
      ostream_iterator<char> stream_it (stream);
      copy (begin_iter, end_iter, stream_it);
      if (read_data) {
        handler->handle_completed_read(stream.str());
      } else {
	ACE_Reactor::instance ()->mask_ops(this,
					   ACE_Event_Handler::READ_MASK,
					   ACE_Reactor::ADD_MASK);
      }
    } else {
      mb->release();
      break;
    }
    mb->release();
  }
  if (DEBUG) std::cout << "Shutting down IOManager." << std::endl;
  shut_down = true;
  return 0;
}

  
bool IOManager::write_block (AsyncIOSvcHandler *handler,
                             ACE_Message_Block* mb) {
  // See if this handler was enqueue for writing purposes,
  // i.e. its queue is non-empty.
  WriteMap::iterator iter = write_map.find(mb);
  if (iter == write_map.end()) {
    return false;
  }
  write_map.erase(iter);

  ACE_Message_Block *write_mb;
  ACE_Time_Value tv(2);
  while(handler->getq(write_mb, &tv) > -1) {
    string test = write_mb->rd_ptr();

    if (test == REMOVE_HANDLER) {
      handler->peer().close();
    } else {
      osock_stream_iterator<char> send_sock_iterator (handler->peer());
      copy(write_mb->rd_ptr(), write_mb->end(), send_sock_iterator);
    } 
    if (test == CLIENT_QUIT_MSG || test == SERVER_QUIT_MSG) {
       handler->sent_act = true;
    }
    write_mb->release();

  }
  return true;
}
